package main

import (
	"io"
	"os"
	"os/signal"
	"strings"
	"sync"
	"syscall"
	"time"

	log "github.com/Sirupsen/logrus"
)

type MessageList struct {
	messages []*Message
	lock     sync.RWMutex
}

func NewMessageList() *MessageList {
	lst := &MessageList{
		messages: make([]*Message, 0),
	}
	return lst
}

func (l *MessageList) Add(m *Message) {
	l.lock.Lock()
	defer l.lock.Unlock()
	l.messages = append(l.messages, m)
}

func (l *MessageList) Iter() <-chan *Message {
	c := make(chan *Message)
	go func() {
		l.lock.RLock()
		for _, m := range l.messages {
			c <- m
		}
		close(c)
		l.lock.RUnlock()
	}()
	return c
}

func (l *MessageList) Empty() {
	l.lock.Lock()
	defer l.lock.Unlock()
	l.messages = l.messages[:0]
}

func (l *MessageList) Len() int {
	l.lock.RLock()
	defer l.lock.RUnlock()
	return len(l.messages)
}

type Worker struct {
	msgConn    *Connection
	jobHandler *JobHandler
	zkaddrs    []string
	zkclient   *ZKClient
	serverAddr string
	stopChan   chan bool
	recvJobCnt int

	connected    bool
	connLock     sync.Mutex
	timeoutTimer *time.Timer
}

func NewWorker() (*Worker, error) {
	worker := &Worker{
		serverAddr: Conf.HostAddr,
		zkaddrs:    Conf.ZkAddrs,
		stopChan:   make(chan bool, 1),
	}
	//从zk中获取server的地址
	if worker.zkaddrs != nil {
		client, err := NewZKClient(worker.zkaddrs)
		if err != nil {
			log.Fatalf("new zk client: %s", err.Error())
		}
		for {
			worker.zkclient = client
			exists, _, err := client.conn.Exists(Conf.ZkIpAndTimeNode)
			if err != nil {
				log.Errorf("zk exists: %s", err.Error())
				time.Sleep(time.Second * 10)
				continue
			}
			if exists {
				iptime, _, err := client.conn.Get(Conf.ZkIpAndTimeNode)
				if err != nil {
					log.Errorf("zk get: %s", err.Error())
					time.Sleep(time.Second * 10)
					continue
				}
				addr := strings.Split(string(iptime), ",")[0]
				worker.serverAddr = addr
			} else {
				log.Error("zk node path not exists")
				time.Sleep(time.Second * 10)
				continue
			}
			break
		}
	}

	for {
		err := worker.Connect(worker.serverAddr)
		if err != nil {
			log.Errorf("connect server %s: %s", worker.serverAddr, err.Error())
			time.Sleep(time.Second * 60)
			iptime, _, err := worker.zkclient.conn.Get(Conf.ZkIpAndTimeNode)
			if err != nil {
				log.Errorf("zk get %s: %s", Conf.ZkIpAndTimeNode, err.Error())
				continue
			}
			addr := strings.Split(string(iptime), ",")[0]
			worker.serverAddr = addr
			continue
		}
		break
	}
	worker.timeoutTimer = time.NewTimer(Conf.RecvJobTimeout.Duration)
	handler := NewJobHandler(worker, Conf.ConcurrencyHandler)
	worker.jobHandler = handler
	log.Infof("connect to %s", worker.serverAddr)
	return worker, nil
}

func (this *Worker) Connect(addr string) error {
	msgConn, err := NewConnection(this.serverAddr)
	if err != nil {
		return err
	}
	this.msgConn = msgConn
	this.connected = true
	return nil
}

func (this *Worker) reConnect() error {
	this.connLock.Lock()
	defer this.connLock.Unlock()
	if this.recvJobCnt == 0 {
		time.Sleep(Conf.NoJobSleepTime.Duration)
	}
	if !this.connected {
		for {
			iptime, _, err := this.zkclient.conn.Get(Conf.ZkIpAndTimeNode)
			if err != nil {
				log.Errorf("reconnect zk get: %s", err.Error())
				time.Sleep(time.Second * 10)
				continue
			}
			addr := strings.Split(string(iptime), ",")[0]
			this.serverAddr = addr
			err = this.Connect(this.serverAddr)
			if err != nil {
				log.Errorf("reconnect error: %s", err.Error())
				time.Sleep(time.Second * 5)
				continue
			}
			break
		}
		log.Infof("reconnect to %s", this.serverAddr)
		err := this.RegisterIdent()
		if err != nil {
			log.Errorf("worker register ident error: %s", err.Error())
		}
		go this.HandleMessage()
		this.timeoutTimer.Reset(Conf.RecvJobTimeout.Duration)
	}
	return nil
}

// 注册worker
func (this *Worker) RegisterIdent() error {
	var jobid string
	if this.jobHandler.curJob != nil {
		jobid = this.jobHandler.curJob.Get(CONF_JOB_ID)
	}
	ident := &ExecutorIdent{
		Id:               Conf.ExecuterId,
		MaxConcurrentJob: Conf.ConcurrencyHandler,
		IDC:              os.Getenv("DC"),
		UUID:             os.Getenv("UUID"),
		JobID:            jobid,
		JobConf:          this.jobHandler.curJob,
		SendIndex:        lastSendSuccessIndex,
	}
	msg := &Message{
		Type: MT_EXECUTOR_IDENT,
		Body: ident,
	}
	return this.FlushMessage(msg)
}

func (this *Worker) HandleHeartbeat() {
	ticker := time.NewTicker(Conf.HeartBeatInterval.Duration)
	for {
		<-ticker.C
		loadavg := GetWorkerLoadAvg()
		addr := this.msgConn.conn.LocalAddr().String()
		ip := strings.Split(addr, ":")[0]
		loadavg.WorkerIp = ip
		msg := &Message{
			Type: MT_HEARTBEAT,
			Body: loadavg,
		}
		log.Debugf("write heart %+v", msg.Body)
		err := this.WriteMessage(msg)
		if err != nil {
			log.Errorf("write heart beat message error: %s", err.Error())
		}
	}
}

func (this *Worker) HandleMessage() {
	for {
		select {
		case <-this.stopChan:
			return
		default:
			msg, err := this.msgConn.ReadMessage()
			if err != nil {
				this.CloseConn()
				if err == io.EOF {
					log.Info("receive eof, will reconnect")
					goto reconnect
				} else {
					log.Errorf("handle message error: %s", err.Error())
					goto reconnect
				}
			}
			log.Infof("receive message: type: %d, %+v", msg.Type, msg.Body)
			switch msg.Type {
			case MT_SUBMIT_JOB:
				log.Infof("receive submit message")
				this.recvJobCnt++
				job := msg.Body.(JobConf)
				this.jobHandler.AddJob(job)
			case MT_KILL_JOB:
				log.Infof("receive kill message")
				job := msg.Body.(JobConf)
				this.jobHandler.KillJobExecute(job)
			case MT_EXIT:
				log.Info("receive exit message")
				this.msgConn.Close()
				this.jobHandler.KillAllJob()
				os.Exit(0)
			case MT_LOG_RESP:
				log.Infof("recevice log resp message")
				select {
				case this.jobHandler.logRespChan <- true:
				case <-time.After(time.Second * 10):
					// 没有程序在 wait response，丢掉resp数据包
				}
			default:
				log.Errorf("unknown message type %d", msg.Type)
			}
		}
	}
reconnect:
	this.reConnect()
}

func (this *Worker) HandleSignal() {
	c := make(chan os.Signal, 1)
	signal.Notify(c, syscall.SIGHUP, syscall.SIGQUIT, syscall.SIGTERM, syscall.SIGINT, syscall.SIGSTOP)
	for {
		s := <-c
		log.Infof("get a signal %s", s.String())
		switch s {
		case syscall.SIGQUIT, syscall.SIGTERM, syscall.SIGSTOP, syscall.SIGINT:
			// this.Stop()
			os.Exit(1)
			return
		case syscall.SIGHUP:
			// TODO reload
			//return
		default:
			return
		}
	}
}

/*
watch zk中server地址的变化，如果有变化，则根据新的地址重连
*/
func (this *Worker) HandleZK() {
	exists, _, err := this.zkclient.conn.Exists(Conf.ZkIpAndTimeNode)
	if err != nil {
		log.Errorf("zk exists error: %s", err.Error())
		return
	}
	if exists {
		for {
			iptime, _, evtChan, err := this.zkclient.conn.GetW(Conf.ZkIpAndTimeNode)
			if err != nil {
				log.Errorf("get zk error: %s", err.Error())
				return
			}
			addr := strings.Split(string(iptime), ",")[0]
			if this.serverAddr != addr {
				log.Infof("zk get server addr %s", addr)
				this.serverAddr = addr
				this.CloseConn()
			}
			evt := <-evtChan
			log.Debugf("receive zk event: %+v", evt)
			// renew zk client. it should never happend in prod
			if evt.Err != nil {
				log.Warnf("zookeeper connection : %s, will renew client", evt.Err.Error())
				this.zkclient.conn.Close()
				for {
					client, err := NewZKClient(this.zkaddrs)
					if err != nil {
						log.Errorf("renew zk client: %s", err.Error())
						time.Sleep(time.Second * 10)
						continue
					}
					this.zkclient = client
					break
				}
			}
		}
	} else {
		log.Errorf("zk node %s not exists", Conf.ZkIpAndTimeNode)
	}
}

func (this *Worker) HandleRecvJobTimeout() {
	<-this.timeoutTimer.C
	if this.recvJobCnt == 0 {
		log.Infof("recv submit job timeout, will exit")
		os.Exit(1)
	}
}

func (this *Worker) Run() {
	go this.HandleHeartbeat()
	if this.zkclient != nil {
		go this.HandleZK()
	}
	go this.jobHandler.HandleJob()
	go this.HandleMessage()
	go this.HandleRecvJobTimeout()
}

func (this *Worker) Stop() {
	this.msgConn.Close()
	this.jobHandler.Stop()
	this.jobHandler.Wait()
	log.Info("stop")
}

// 异步方式发送数据
func (this *Worker) WriteMessage(msg *Message) error {
	return this.msgConn.WriteMessage(msg)
}

// 同步方式发送数据
func (this *Worker) FlushMessage(msg *Message) error {
	err := this.msgConn.FlushMessage(msg)
	if err != nil {
		this.jobHandler.logRespChan <- false
	}
	return nil
}

func (this *Worker) CloseConn() error {
	this.connLock.Lock()
	defer this.connLock.Unlock()
	if this.connected {
		if this.msgConn != nil {
			this.msgConn.Close()
		}
		if this.recvJobCnt == 0 {
			// 在连接关闭的情况下,保证不会超时
			this.timeoutTimer.Reset(time.Hour * 24 * 30)
		}
		time.Sleep(time.Second * 15)
		this.connected = false
	}
	return nil
}
